package com.cryptoneedle.center.source.service.metadata;


import com.bubbles.engine.data.core.repository.BaseRepository;
import com.bubbles.engine.data.core.service.BaseService;
import com.cryptoneedle.center.source.entity.metadata.Catalog;
import com.cryptoneedle.center.source.entity.metadata.Database;
import com.cryptoneedle.center.source.entity.metadata.Table;
import com.cryptoneedle.center.source.key.metadata.TableKey;
import com.cryptoneedle.center.source.repository.metadata.TableRepository;
import com.cryptoneedle.center.source.service.crud.*;
import com.cryptoneedle.center.source.util.JdbcUtil;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>description: 源数据源-表 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@Slf4j
@Service
public class TableService extends BaseService<Table, TableKey> {

    @Resource
    private TableService tableService;
    @Resource
    private TableRepository tableRepository;

    @Override
    public BaseRepository<Table, TableKey> getRepository() {
        return this.tableRepository;
    }

    @Autowired
    private FoundDataService select;
    @Autowired
    private AddDataService add;
    @Autowired
    private PatchDataService patch;
    @Autowired
    private SaveDataService save;
    @Autowired
    private DeleteDataService delete;

    @Transactional
    public void syncDatabase(Catalog catalog, Database database) {
        JdbcTemplate jdbcTemplate = JdbcUtil.getJdbcTemplate(catalog);
        List<Table> tables = jdbcTemplate.query(
                catalog.getDatabaseType().getDatabase().getTableSql(database.getDatabaseName()),
                (resultSet, rowNum) -> {
                    Table table = new Table()
                            .setCatalogName(database.getCatalogName())
                            .setDatabaseName(database.getDatabaseName())
                            .setTableName(resultSet.getString("tableName"))
                            .setComment(resultSet.getString("tableComment"))
                            .setRowCount(resultSet.getLong("statisticRowCount"))
                            .setAvgRowLen(resultSet.getLong("statisticAvgRowLen"))
                            .setEstimateMegaBytes(null)
                            .setStatisticDt(resultSet.getDate("statisticDatetime"))
                            .setActiveDimensionName(null)
                            .setDorisCatalogName(database.getDorisCatalogName())
                            .setSystemCode(database.getDefaultSystemCode())
                            .setEnabled(false)
                            .setTransTableName(null)
                            .setTransTableNameManual(false)
                            .setTransComment(null)
                            .setTransCommentManual(false)
                            .setTransBucketCount(null);

                    // 计算表占用空间 setEstimateMegaBytes
                    calcuteEstimateMegaBytes(table);
                    return table;
                }

        );


        tableService.compareData(database, tables);
    }

    private void calcuteEstimateMegaBytes(Table tabke) {
        Long rowCount = tabke.getRowCount();
        Long avgDataSizeByte = tabke.getAvgRowLen();
        if (rowCount == null || avgDataSizeByte == null) {
            return;
        }
        BigDecimal estimateMegaBytes = BigDecimal.valueOf(rowCount * avgDataSizeByte).divide(BigDecimal.valueOf(1048576), 2, RoundingMode.HALF_UP);
        tabke.setEstimateMegaBytes(estimateMegaBytes);
    }

    @Transactional
    public void compareData(Database database, List<Table> dealList) {
        List<Table> originList = select.tables(database);
        // 已存在
        Map<String, Table> originMap = Maps.uniqueIndex(originList, Table::getTableName);
        // 待处理
        Map<String, Table> dealMap = Maps.uniqueIndex(dealList, Table::getTableName);

        // 新增
        List<Table> extraList = dealList.stream().filter(deal -> !originMap.containsKey(deal.getTableName())).toList();

        // 保存
        List<Table> existsList = originList.stream()
                .filter(origin -> dealMap.containsKey(origin.getTableName()))
                .peek(origin -> {
                    Table deal = dealMap.get(origin.getTableName());
                    origin.setComment(deal.getComment())
                            .setRowCount(deal.getRowCount())
                            .setStatisticDt(deal.getStatisticDt());
                }).toList();

        // 移除
        List<Table> missList = originList.stream().filter(item -> !dealMap.containsKey(item.getTableName())).toList();

        add.tables(extraList);
        save.tables(existsList);
        delete.tables(missList);

        log.info("[SYNC] {} -> {} -> 元数据采集 -> Table              => 共计{} | 新增{} | 删除{}", database.getCatalogName(), database.getDatabaseName(), String.format("%5d", dealList.size()), String.format("%5d", extraList.size()), String.format("%5d", missList.size()) + (
                !missList.isEmpty() ? " => " + missList.stream().map(Table::getTableName).collect(Collectors.joining(",")) : ""));
    }
}